package com.xrui.hbase;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;

import com.xrui.hbase.exception.TableNotFoundException;
import com.xrui.hbase.schema.TableLayout;
import com.xrui.hbase.schema.TableLayoutsParser;
import com.xrui.hbase.security.Kerboros;
import com.xrui.hbase.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HConnectionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;

/**
 * HBase instance class that contains configuration and table information.
 * Multiple instances of HBase can be installed onto a single HBase cluster.
 * This class represents a single one of those instances.
 */
public final class HBase implements TableFactory, ReferenceCountable<HBase> {
    //region static
    private static final Logger LOG = LoggerFactory.getLogger(HBase.class);
    /**
     * Tracks the state of this HBase instance.
     */
    private final AtomicReference<State> mState = new AtomicReference<>(State.UNINITIALIZED);
    /**
     * Retain counter. When decreased to 0, the HBase may be closed and disposed of.
     */
    private final AtomicInteger mRetainCount = new AtomicInteger(0);
    //endregion

    //region Variables
    /**
     * The hadoop configuration.
     */
    private final Configuration mConf;
    /**
     * URI for this HBase instance.
     */
    private final HBaseURI mURI;
    /**
     * A map from table names to their layouts.
     */
    private final Map<String, TableLayout> mTableLayoutCache;




    /**
     * A connection to the HBase
     */
    private HConnection mhConnection;
    /**
     * HBase admin interface. Lazily initialized through {@link #getHBaseAdmin()}.
     */
    private HBaseAdmin mAdmin = null;
    private TableLayoutsParser mTableLayoutsParser;

    //region default construct: only can be instance by calling HBase.open(...)
    HBase(
        HBaseURI hbaseURI,
        Configuration conf
    ) throws IOException {
        // Deep copy the configuration.
        mConf = new Configuration(conf);

        // Validate arguments.
        mURI = Preconditions.checkNotNull(hbaseURI);

        // Configure the ZooKeeper quorum:
        mConf.setStrings("hbase.zookeeper.quorum", mURI.getZookeeperQuorum().toArray(new String[0]));
        mConf.setInt("hbase.zookeeper.property.clientPort", mURI.getZookeeperClientPort());


        Kerboros.login(getConfiguration());

        // Check for an instance name.
        Preconditions.checkArgument(mURI.getInstance() != null,
            "HBaseURI '%s' does not specify a HBase instance name.", mURI);

        if (LOG.isDebugEnabled()) {
            Debug.logConfiguration(mConf);
            LOG.debug("Opening HBase instance '{}'", mURI);
        }
        LOG.debug("HBase instance '{}' is now opened.", mURI);

        mTableLayoutCache = new HashMap<>();
        mTableLayoutsParser = TableLayoutsParser.newParser(
                Resources.openSystemTextResource("hbase_table_schema.json")
        );

        // Make sure the data version for the client matches the cluster.
        LOG.debug("Validating version for HBase instance '{}'.", mURI);

        mRetainCount.set(1);
        final State oldState = mState.getAndSet(State.OPEN);
        Preconditions.checkState(oldState == State.UNINITIALIZED,
            "Cannot open HBase instance in state %s.", oldState);

        DebugResourceTracker.get().registerResource(this);
    }

    /**
     * Opens a HBase instance by URI.
     * <p>
     * Caller does not need to call {@code HBase.retain()}, but must call {@code HBase.release()}
     * after the returned {@code HBase} will no longer be used.
     *
     * @param uri URI specifying the HBase instance to open.
     * @return the specified HBase instance.
     * @throws IOException on I/O error.
     */
    public static HBase open(HBaseURI uri) throws IOException {
        return HBaseFactory.get().open(uri);
    }

    //endregion

    /**
     * Opens a HBase instance by URI.
     *
     * @param uri  URI specifying the HBase instance to open.
     * @param conf Hadoop configuration.
     * @return the specified HBase instance.
     * @throws IOException on I/O error.
     */
    public static HBase open(HBaseURI uri, Configuration conf) throws IOException {
        return HBaseFactory.get().open(uri, conf);
    }

    //endregion

    //region APIs

    /**
     * Opens a HBaseTable by name.
     * <p>
     * <p>Clients must take care to release the HBaseTable instance when finished.</p>
     *
     * @param tableName Name of the table to open.
     * @return the opened HBase table.
     * @throws IOException on I/O error.
     */
    @Override
    public HBaseTable openTable(String tableName) throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot open table in HBase instance %s in state %s.", this, state);

        if (null == mhConnection) {
            mhConnection = HConnectionManager.createConnection(mConf);
        }

//        if (!getTableNames().contains(tableName)) {
//            throw new TableNotFoundException(
//                HBaseURI.newBuilder(mURI).withTableName(tableName).build());
//        }

        return new HBaseTable(
            this,
            tableName,
            mhConnection);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public HBase retain() {
        LOG.debug("Retaining {}.", this);
        final int counter = mRetainCount.getAndIncrement();
        Preconditions.checkState(counter >= 1,
            "Cannot retain HBase instance %s: retain counter was %s.", this, counter);
        return this;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void release() throws IOException {
        LOG.debug("Releasing {}", this);
        final int counter = mRetainCount.decrementAndGet();
        Preconditions.checkState(counter >= 0,
            "Cannot release HBase instance %s: retain counter is now %s.", this, counter);
        if (counter == 0) {
            close();
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object obj) {
        if (null == obj) {
            return false;
        }
        if (obj == this) {
            return true;
        }
        if (!getClass().equals(obj.getClass())) {
            return false;
        }
        final HBase other = (HBase) obj;

        // Equal if the two instances have the same URI:
        return mURI.equals(other.getURI());
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return mURI.hashCode();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(HBase.class)
            .add("id", System.identityHashCode(this))
            .add("uri", mURI)
            .add("retain-count", mRetainCount)
            .add("state", mState.get())
            .toString();
    }

    public HBaseURI getURI() {
        return mURI;
    }

    public TableLayout getTableLayout(HBaseURI uri) {
        //TODO: So far, getTableLayout(HBaseURI uri) only support Table Name with no prefix
        String table = uri.getTable();
        Preconditions.checkNotNull(table);

        if (!mTableLayoutCache.containsKey(table)) {
            TableLayout tableLayout = mTableLayoutsParser.get(table, mConf,this);
            mTableLayoutCache.put(table, tableLayout);
        }

        return mTableLayoutCache.get(table);
    }

    /**
     * Gets the current HBaseAdmin instance for this HBase. This method will open a new
     * HBaseAdmin if one doesn't exist already.
     *
     * @return The current HBaseAdmin instance for this HBase.
     * @throws IOException If there is an error opening the HBaseAdmin.
     */
    public synchronized HBaseAdmin getHBaseAdmin() throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot create HBase admin for HBase instance %s in state %s.", this, state);
        if (null == mAdmin) {
            mAdmin = new HBaseAdmin(mConf);
        }
        return mAdmin;
    }

    public List<String> getTableNames() throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
            "Cannot create table names in HBase instance %s in state %s.", this, state);

        //HTableInterface test = getConnection().getTable("Spark_RuleChange");
        //test.getScanner(new byte[] {'c'});

        HTableDescriptor[] tableDescriptors = getHBaseAdmin().listTables();
        List<String> tables = new ArrayList<>();
        for (HTableDescriptor tableDescriptor : tableDescriptors) {
            tables.add(tableDescriptor.getTableName().toString());
        }
        return tables;
    }

    public List<String> getTableNames(Pattern pattern) throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
                "Cannot create table names in HBase instance %s in state %s.", this, state);

        //HTableInterface test = getConnection().getTable("Spark_RuleChange");
        //test.getScanner(new byte[] {'c'});

        HTableDescriptor[] tableDescriptors = getHBaseAdmin().listTables(pattern);
        List<String> tables = new ArrayList<>();
        for (HTableDescriptor tableDescriptor : tableDescriptors) {
            tables.add(tableDescriptor.getTableName().toString());
        }
        return tables;
    }

    public TableName getDefaultNameSpaceAndPrefix() throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
                "Cannot create table names in HBase instance %s in state %s.", this, state);

        //HTableInterface test = getConnection().getTable("Spark_RuleChange");
        //test.getScanner(new byte[] {'c'});

        List<String> tables =  getTableNames(Pattern.compile(".*_Type"));

        return TableName.valueOf(tables.get(0));
    }


    public List<String> getCdfTableNames(String namespace) throws IOException {
        final State state = mState.get();
        Preconditions.checkState(state == State.OPEN,
                "Cannot create table names in HBase instance %s in state %s.", this, state);

        //HTableInterface test = getConnection().getTable("Spark_RuleChange");
        //test.getScanner(new byte[] {'c'});
        HTableDescriptor[] tableDescriptors = getHBaseAdmin().listTableDescriptorsByNamespace(namespace);


        List<String> tables = new ArrayList<>();
        for (HTableDescriptor tableDescriptor : tableDescriptors) {
            tables.add(tableDescriptor.getTableName().toString());
        }
        return tables;
    }

    //region private methods
    public Configuration getConfiguration() {
        return mConf;
    }

    //endregion

    /**
     * Releases all the resources used by this HBase instance.
     *
     * @throws IOException on I/O error.
     */
    private void close() throws IOException {
        final State oldState = mState.getAndSet(State.CLOSED);
        Preconditions.checkState(oldState == State.OPEN || oldState == State.UNINITIALIZED,
            "Cannot close HBase instance %s in state %s.", this, oldState);

        LOG.debug("Closing {}.", this);

        synchronized (this) {
            ResourceUtils.closeOrLog(mAdmin);
            mAdmin = null;
            ResourceUtils.closeOrLog(mhConnection);
        }

        if (oldState != State.UNINITIALIZED) {
            DebugResourceTracker.get().unregisterResource(this);
        }
        LOG.debug("{} closed.", this);
    }

    /**
     * States of a HBase instance.
     */
    private enum State {
        /**
         * Initialization begun but not completed.  Retain counter and DebugResourceTracker counters
         * have not been incremented yet.
         */
        UNINITIALIZED,
        /**
         * Finished initialization.  Both retain counters and DebugResourceTracker counters have been
         * incremented.  Resources are successfully opened and this HBase's methods may be used.
         */
        OPEN,
        /**
         * Closed.  Other methods are no longer supported.  Resources and connections have been closed.
         */
        CLOSED
    }
    //endregion
}
